package com.snail.starter.kafka.abs;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.snail.starter.utils.id.ObjectId;
import com.snail.starter.utils.json.JsonUtils;
import com.snail.starter.utils.lambda.GenericEntityBuilder;
import com.snail.starter.commons.threadlocal.UserContext;
import com.snail.starter.commons.threadlocal.UserContextUtils;
import com.snail.starter.kafka.KafkaListener;
import com.snail.starter.mq.dto.MQMessage;
import com.snail.starter.mq.enums.LocalMessageConsumerStatusEnum;
import com.snail.starter.mq.feign.request.LocalMessageConsumerRequest;
import com.snail.starter.mq.feign.response.LocalMessageProducerResponse;
import com.snail.starter.mq.rpc.LocalMessageRpc;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.MDC;
import org.springframework.context.ApplicationContext;
import org.springframework.kafka.listener.BatchMessageListener;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author: snail framework
 * @Date: 2023/1/31 下午3:47
 * @Version: 1.0.0
 */
@Slf4j
public abstract class AbstractKafkaBatchMessageListener implements BatchMessageListener<String, String>, KafkaListener {

    @Resource
    LocalMessageRpc localMessageRpc;

    @Resource
    ApplicationContext applicationContext;

    @Resource
    ObjectMapper objectMapper;

    public Long before(MQMessage mqMessage) {
        String                       traceId     = ObjectId.nextId();
        UserContext userContext = UserContextUtils.get();
        userContext.setTenantId(mqMessage.getTenantId());
        userContext.setTraceId(traceId);
        userContext.setSpanId(traceId);
        MDC.put("traceId", traceId);
        MDC.put("spanId", traceId);
        MDC.put("parentId", traceId);
        UserContextUtils.set(userContext);
        LocalMessageProducerResponse messageResponse = localMessageRpc.getOneByMessageIdProducer(mqMessage.getMessageId());
        if (null == messageResponse) {
            messageResponse = new LocalMessageProducerResponse();
        }
        return localMessageRpc.addByConsumer(GenericEntityBuilder.of(LocalMessageConsumerRequest::new)
                .with(LocalMessageConsumerRequest::setTenantId, mqMessage.getTenantId())
                .with(LocalMessageConsumerRequest::setMessageId, mqMessage.getMessageId())
                .with(LocalMessageConsumerRequest::setTopic, topic())
                .with(LocalMessageConsumerRequest::setGroup, group())
                .with(LocalMessageConsumerRequest::setServerName, applicationContext.getApplicationName())
                .with(LocalMessageConsumerRequest::setBizId, messageResponse.getBizId())
                .with(LocalMessageConsumerRequest::setBizCode, messageResponse.getBizCode())
                .with(LocalMessageConsumerRequest::setMessageType, messageResponse.getMessageType())
                .with(LocalMessageConsumerRequest::setContent, mqMessage.getContent())
                .with(LocalMessageConsumerRequest::setTraceId, userContext.getTraceId())
                .with(LocalMessageConsumerRequest::setSpanId, userContext.getSpanId())
                .build());
    }

    public abstract boolean onMessage(String content);

    @Override
    public void onMessage(List<ConsumerRecord<String, String>> consumerRecords) {
        if (null == consumerRecords || consumerRecords.isEmpty()) {
            return;
        }
        consumerRecords.forEach(consumerRecord -> {
            String message = consumerRecord.value();
            if (null == message || "".equals(message)) {
                return;
            }
            MQMessage mqMessage = JsonUtils.toBean(message, MQMessage.class);
            log.info("KafkaMessageListener#onMessage,topic:{}, group:{}, message:{}", topic(), group(), message);
            if (null == mqMessage || null == mqMessage.getContent() || "".equals(mqMessage.getContent())) {
                if (!onMessage(message)) {
                    log.error("old style consumer message error,topic:{}, group:{}, message:{}", topic(), group(), message);
                }
                return;
            }
            Long consumerMessageId = null;
            try {
                consumerMessageId = before(mqMessage);
            } catch (Throwable throwable) {
                log.error("", throwable);
                log.error("KafkaMessageListener#before message:{}", message);
            }
            LocalMessageConsumerStatusEnum consumerStatus = LocalMessageConsumerStatusEnum.SUCCEED;
            if (!onMessage(mqMessage.getContent())) {
                log.error("new style consumer message error,topic:{}, group:{}, message:{}", topic(), group(), message);
                consumerStatus = LocalMessageConsumerStatusEnum.FAILED;
            }
            if (null == consumerMessageId) {
                return;
            }
            if (!after(mqMessage.getTenantId(), consumerMessageId, consumerStatus)) {
                log.error("local message modify status fail,consumerMessageId:{},consumerStatus:{}", consumerMessageId, consumerStatus);
            }
        });
    }

    public boolean after(Long tenantId, Long consumerMessageId, LocalMessageConsumerStatusEnum consumerStatus) {
        return localMessageRpc.modifyStatusByConsumerTenantId(tenantId, consumerMessageId, consumerStatus.getStatus());
    }

}